#include "postgresql_connection.h"
#include <sstream>
#include "SQLcommand_impl.h"
#include "database_module_impl.h"
#include "postgresql_query_result.h"
#ifdef WIN32
#include <Windows.h>
#endif


#ifdef ENABLE_POSTGRESQL
#pragma comment(lib, "libpq.lib")

namespace utility
{
namespace database
{
    postgresql_connection::postgresql_connection(uint32 index, database_module_impl * db_module): 
        _index(index), _connected(false), _db_module(db_module), 
        _worker_thread(NULL), _terminated(true), _pg_conn(NULL)
    {
        db_module->register_runnable_object(this);
    }

    postgresql_connection::~postgresql_connection()
    {
        _db_module->unregister_runnable_object(this);
        disconnect();
    }

    bool postgresql_connection::connect(char * host, char * user, char * password, char * db_name, uint32 port)
    {
        disconnect();

        std::ostringstream strm_port;
        strm_port << port;
        _pg_conn = PQsetdbLogin(host,
            strm_port.str().c_str(),
            NULL,
            NULL,
            db_name,
            user,
            password);

        if (PQstatus(_pg_conn) == CONNECTION_BAD)
        {
            PQfinish(_pg_conn);
            return false;
        }
        else
        {
            _terminated = false;
            _worker_thread = new boost::thread(boost::bind(&postgresql_connection::_start_work, this));
            return true;
        }
    }

    void postgresql_connection::disconnect()
    {
        _terminated = true;
        if (_worker_thread)
        {
            _worker_thread->join();
            delete _worker_thread;
            _worker_thread = NULL;
        }

        if (_pg_conn)
        {
            PQfinish(_pg_conn);
            _pg_conn = NULL;
        }

        _connected = false;
    }

    SQLcommand_point postgresql_connection::create_command(char * SQLtext)
    {
        return SQLcommand_point(new SQLcommand_impl(SQLtext, this));
    }

    void postgresql_connection::async_execute(SQLcommand_point command)
    {
        boost::lock_guard<lock_type> guard(_lock);
        _waiting_commands.push_back(command);
    }

    uint32 postgresql_connection::escape_string(const char * src, size_t src_length, char * dst)
    {
        int err;
        size_t escaped_len;

        escaped_len = PQescapeStringConn(_pg_conn, dst, src, src_length, &err);
        if (err != 0)
        {
            throw error_code(3000, "error occurred when escaping string");
        }
    }

    bool postgresql_connection::run()
    {
        if (_completed_commands.size() > 0)
        {
            boost::lock_guard<lock_type> guard(_lock);
            completed_commands::const_iterator i = _completed_commands.begin();
            while (i!= _completed_commands.end())
            {
                completed_commands::const_iterator j = i++;
                const completed_item & item = *j;
                (item.first)->get_callback_function()->query_callback(
                    item.second._result, 
                    item.second._error
                    );
                _completed_commands.erase(j);
            }
        }
        return false;
    }

    void postgresql_connection::_start_work()
    {
        while(!_terminated)
        {
            if (_waiting_commands.size() > 0)
            {
                boost::lock_guard<lock_type> guard(_lock);
                waiting_commands::iterator i = _waiting_commands.begin();
                while (i != _waiting_commands.end())
                {
                    waiting_commands::iterator j = i++;
                    SQLcommand_point command = *j;
                    if (SQLcommand::QUERY == command->get_type())
                    {
                        _execute_query(command);
                        _waiting_commands.erase(j);
                    }
                    else if (SQLcommand::NON_QUERY == command->get_type())
                    {
                        _execute_non_query(command);
                        _waiting_commands.erase(j);
                    }

                }
            }  
            Sleep(1);
        }
        //printf("exit thread\n");
    }

    void postgresql_connection::_execute_query(SQLcommand_point command)
    {
        completed_item item;
        item.first = command;

        try
        {
            std::string SQL = command->get_SQL();

            PGresult * result =PQexec(_pg_conn,SQL.c_str());

            if (PQresultStatus(result) != PGRES_TUPLES_OK)
            {
                item.second._error.set_error(1, PQresultErrorMessage(result));
            }
            else
            {
                item.second._result = new postgresql_query_result(result);
            }
            PQclear(result);
        }
        catch(error_code & ec)
        {
            item.second._error = ec;
        }

        _completed_commands.push_back(item);
    }

    void postgresql_connection::_execute_non_query(SQLcommand_point command)
    {
        completed_item item;
        item.first = command;

        try
        {
            std::string SQL = command->get_SQL();
            PGresult * result =PQexec(_pg_conn,SQL.c_str());
            if (PQresultStatus(result) != PGRES_COMMAND_OK)
            {
                item.second._error.set_error(1, PQresultErrorMessage(result));
            }
            PQclear(result);
        }
        catch(error_code & ec)
        {
            item.second._error = ec;
        }

        _completed_commands.push_back(item);
    }
}
}

#endif // endif of ENABLE_POSTGRESQL

